package com.bw.wjw.function;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class LogSplitProcessFunction extends ProcessFunction<String, String> {

     private OutputTag<String> errorTag = null ;
     private OutputTag<String> startTag = null ;
     private OutputTag<String> displayTag = null ;
     private OutputTag<String> actionTag = null ;

     /**
     * 构造方法
     */
    public LogSplitProcessFunction(
            OutputTag<String> errorTag,
            OutputTag<String> startTag,
            OutputTag<String> displayTag,
            OutputTag<String> actionTag){
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }


    @Override
    public void processElement(String value,
                               ProcessFunction<String, String>.Context ctx,
                               Collector<String> out) throws Exception {
        // 解析JSON
        JSONObject jsonObject = JSON.parseObject(value);

        // a. 错误日志
        String errorValue = jsonObject.getString("err");
        if (null != errorValue) {
            ctx.output(errorTag, value);
        }
        jsonObject.remove("err");

        // b. 启动日志
        String startValue = jsonObject.getString("start");
        if (null != startValue) {
            ctx.output(startTag, value);
        }

        // c. 页面日志
        String pageValue = jsonObject.getString("page");
        if (null != pageValue) {
            // 曝光日志
            JSONArray displaysArray = jsonObject.getJSONArray("displays");
            // 动作日志
            JSONArray actionsArray = jsonObject.getJSONArray("actions");

            jsonObject.remove("displays");
            jsonObject.remove("actions");
            out.collect(jsonObject.toJSONString());

            // 曝光日志：displays 拆分为 多个display
            if (null != displaysArray && !displaysArray.isEmpty()) {
                // 遍历每个display曝光日志数据
                for(int index = 0; index < displaysArray.size(); index ++){
                    String displayValue = displaysArray.getString(index);
                    // 添加字段：display和值displayValue
                    jsonObject.put("display", JSONObject.parseObject(displayValue));
                    // 侧边流输出
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }

            jsonObject.remove("display");
            // 动作日志：actions 拆分为 多个action
            if (null != actionsArray && !actionsArray.isEmpty()) {
                // 遍历每个action动作日志数据
                for(int index = 0; index < actionsArray.size(); index ++){
                    String actionValue = actionsArray.getString(index);
                    // 添加字段：action和值actionValue
                    jsonObject.put("action", JSONObject.parseObject(actionValue));
                    // 侧边流输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }
}
